package com.study.chapter05.operator.sink;

import com.study.entity.Event;
import com.study.chapter05.source.ClickSource;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/7/30 17:16
 */
public class Sink2ElasticSearchTest {
    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("192.168.300.100",9200,"http"));

        //创建一个ElasticsearchSinkFunction
        ElasticsearchSinkFunction<Event> esSinkFunction = new ElasticsearchSinkFunction<Event>() {

            @Override
            public void process(Event event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                Map<String, String> data = new HashMap<>();
                data.put(event.user, event.url);

                IndexRequest request = Requests.indexRequest()
                        .index("clicks")
                        .type("type") //若是ES 6则要定义type
                        .source(data);
                requestIndexer.add(request);
            }
        };

        //将数据写入到es
        ElasticsearchSink<Event> esSink = new ElasticsearchSink.Builder<Event>(httpHosts, esSinkFunction).build();
        stream.addSink(esSink);

        stream.print();

        env.execute();
    }
}
